package two.phase.termination;

import guarded.suspension.AlarmAgent;
import guarded.suspension.AlarmInfo;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Administrator on 2017/3/16.
 */
public class AlarmSendingThread extends AbstractTerminationThread {


    private final AlarmAgent alarmAgent = new AlarmAgent() ;

    private final BlockingQueue<AlarmInfo> alarmQueue;

    private final ConcurrentMap<String,AtomicInteger> submittedAlarmRegistry;

    public AlarmSendingThread() {
        this.alarmQueue = new ArrayBlockingQueue<AlarmInfo>(100);
        this.submittedAlarmRegistry = new ConcurrentHashMap<String, AtomicInteger>();
        alarmAgent.init();
    }

    @Override
    protected void doRun() throws Exception {
        AlarmInfo alarm;

        alarm=alarmQueue.take();

        terminationToken.reservations.decrementAndGet();

        try {
            alarmAgent.sendAlarm(alarm);
        }catch (Exception e){
            e.printStackTrace();
        }

        if ( AlarmType.RESUME == alarm.getType() ){
            String key = AlarmType.FAULT.toString() + ":" + alarm.getId() + "@" +alarm.getExtraInfo() ;
            submittedAlarmRegistry.remove(key);

            key = AlarmType.RESUME.toString() + ":" + alarm.getId() + "@" +alarm.getExtraInfo() ;
            submittedAlarmRegistry.remove(key);
        }

    }

    public int sendAlarm(final AlarmInfo alarmInfo){

        AlarmType type=alarmInfo.getType();
        String id=alarmInfo.getId();
        String extraInfo=alarmInfo.getExtraInfo();

        if(terminationToken.isToShutdown()){
            System.err.println("rejected alarm : "+ id + "," + extraInfo);
            return -1;
        }

        int duplicateSubmissionCount = 0;

        try {
            AtomicInteger prevSubmittedCounter;
            String key = type.toString() + ":" + alarmInfo.getId() + "@" +alarmInfo.getExtraInfo() ;
            prevSubmittedCounter = submittedAlarmRegistry.putIfAbsent(key,new AtomicInteger(0));
            if(null == prevSubmittedCounter){
                terminationToken.reservations.incrementAndGet();
                alarmQueue.put(alarmInfo);
            }else{
                //说明map中存在该警告，也就是说，该警报没有解除，所以只是把该警报的计数+1；
                duplicateSubmissionCount=prevSubmittedCounter.incrementAndGet();
            }
        }catch (Throwable t){
            t.printStackTrace();
        }
        return duplicateSubmissionCount;
    }

    @Override
    protected void doCleanup(Exception cause) {
        if(null != cause && !(cause instanceof InterruptedException)){
            cause.printStackTrace();
        }
        alarmAgent.disconnect();
    }
}

